package com.weiyu.hbase.utils;


import com.weiyu.hbase.config.HbaseConfig;
import com.weiyu.hbase.entity.HBaseColumn;
import com.weiyu.hbase.entity.HBaseColumnFamily;
import com.weiyu.hbase.entity.HBaseRow;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.client.coprocessor.AggregationClient;
import org.apache.hadoop.hbase.client.coprocessor.LongColumnInterpreter;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.DependsOn;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StopWatch;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;


@DependsOn("springContextHolder")		//控制依赖顺序，保证springContextHolder类在之前已经加载
@Component
public class HBaseUtils {
    private Logger logger = LoggerFactory.getLogger(this.getClass());

    /**
     * 手动获取hbaseConfig配置类对象
     */
     private static HbaseConfig con = SpringContextHolder.getBean("hbaseConfig");

    private static Configuration conf = HBaseConfiguration.create();

   // private static ExecutorService pool = Executors.newScheduledThreadPool(20);
   /**
    * 设置连接池
    */
    private static ThreadPoolExecutor pool = new ThreadPoolExecutor(20, 1000, 0, TimeUnit.SECONDS, new LinkedBlockingQueue<>());
    private static Connection connection = null;
    private static HBaseUtils instance = null;
    private static Admin admin = null;

    /**
     * 简单单例模式，私有化构造方法
     */
    private HBaseUtils() {
        if (connection == null) {
            try {
                connection = ConnectionFactory.createConnection(con.configuration(), pool);
                admin = connection.getAdmin();
            } catch (IOException e) {
                logger.error("HbaseUtils实例初始化失败！错误信息为：" + e.getMessage(), e);
            }
        }
    }

    /**
     * 简单单例方法（懒汉），如果autowired自动注入就不需要此方法
     */
    public static synchronized HBaseUtils getInstance() {
        if (instance == null) {
            instance = new HBaseUtils();
        }
        return instance;
    }

    /**
     *     基础概念
     *     NamespaceDescriptor:维护命名空间的信息,但是namespace，一般用shell来建立
	 *     Admin:提供了一个接口来管理 HBase 数据库的表信息
	 *     HTableDescriptor:维护了表的名字及其对应表的列族,通过HTableDescriptor对象设置表的特性
	 *     HColumnDescriptor:维护着关于列族的信息,可以通过HColumnDescriptor对象设置列族的特性
     */


    /**
     * @description  创建表
     * @param tableName
     * @param columnFamily
     * @return void
     * @author abner chow
     * @date 2019/4/5 16:25
     */
    public void createTable(String tableName, String[] columnFamily) throws IOException {
        TableName name = TableName.valueOf(tableName);
        //如果存在则删除
        if (admin.tableExists(name)) {
            logger.error("create htable error! this table {} already exists!", name);
        } else {
            HTableDescriptor desc = new HTableDescriptor(name);
            for (String cf : columnFamily) {
                desc.addFamily(new HColumnDescriptor(cf));
            }
            admin.createTable(desc);
        }
    }


    /**
     * @description 判断表格是否存在
     * @param tableName  表名
     * @return boolean
     * @author abner chow
     * @date 2019/4/12 17:56
     */
    public static boolean isExistTable(String tableName) throws IOException {
        return admin.tableExists(TableName.valueOf(tableName));
    }


    /**
     * @description 判断列族是否存在
     * @param tableName
     * @param columnFamily
     * @return boolean
     * @author abner chow
     * @date 2019/5/5 11:04
     */
    public  boolean isExistFamily(String tableName,String columnFamily ) throws IOException {
        TableName tablename = TableName.valueOf(tableName);
        if (admin.tableExists(tablename)) {
            HTableDescriptor hTableDescriptor = admin.getTableDescriptor(tablename);
            return hTableDescriptor.hasFamily(Bytes.toBytes(columnFamily));
        } else {
            logger.error("this table: " + tableName + " does not exist");
        }
        return Boolean.FALSE;
    }


    /**
     * @description 添加列族
     * @param tableName     表名
     * @param columnFamily  列族
     * @return void
     * @author abner chow
     * @date 2019/4/5 16:35
     */
    public void addColumnFamily(String tableName,String columnFamily) throws IOException {
        TableName tablename = TableName.valueOf(tableName);
        if (admin.tableExists(tablename)) {
            admin.disableTable(tablename);
            HTableDescriptor hTableDescriptor = admin.getTableDescriptor(tablename);
            HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(columnFamily);
            hTableDescriptor.addFamily(hColumnDescriptor);
            admin.modifyTable(tablename, hTableDescriptor);
            admin.enableTable(tablename);
        } else {
            logger.error("this table: " + tableName + " does not exist");
        }
    }


    /**
     * 插入记录（单行单列族-多列多值）
     *
     * @param tableName         表名
     * @param row               行名
     * @param columnFamilys     列族名
     * @param columns           列名（数组）
     * @param values            值（数组）（且需要和列一一对应）
     */
    public void insertRecords(String tableName, String row, String columnFamilys, String[] columns, String[] values) throws IOException {
        TableName name = TableName.valueOf(tableName);
        Table table = connection.getTable(name);
        Put put = new Put(Bytes.toBytes(row));
        for (int i = 0; i < columns.length; i++) {
            put.addColumn(Bytes.toBytes(columnFamilys), Bytes.toBytes(columns[i]), Bytes.toBytes(values[i]));
            table.put(put);
        }
    }

    /**
     * @description 插入记录（单行单列族-单列单值）
     * @param tableName     表名
     * @param rowName       行名
     * @param columnFamily  列族名
     * @param column        列名
     * @param value         值
     * @return void
     * @author abner chow
     * @date 2019/4/5 16:38
     */
    public void insertOneRecord(String tableName, String rowName, String columnFamily, String column, String value) throws IOException {
        TableName name = TableName.valueOf(tableName);
        Table table = connection.getTable(name);
        Put put = new Put(Bytes.toBytes(rowName));
        put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));
        table.put(put);
    }


    /**
     * @description 按存储结构批量存入
     * @Param: [hBaseRow]
     * @Return: void
     * @Author: abner chow
     * @Date: 2020/4/8 15:44
     */
    public  void insert(HBaseRow hBaseRow) throws IOException {
        String tableName = hBaseRow.getTableName();
        List<HBaseColumnFamily> hBaseColumnFamilies = hBaseRow.getHBaseColumnFamilies();
        if (!CollectionUtils.isEmpty(hBaseColumnFamilies)) {
            for (HBaseColumnFamily hBaseColumnFamily : hBaseColumnFamilies) {
                String columnFamilyName = hBaseColumnFamily.getColumnFamilyName();
                if (!isExistTable(tableName)) {
                    //获取列族名
                    String[] nameAry = {columnFamilyName};
                    //不存在则创建表
                    createTable(tableName, nameAry);
                } else {
                    Boolean isExistFamily = isExistFamily(tableName, columnFamilyName);
                    if (!isExistFamily) {
                        //不存在添加列族
                        addColumnFamily(tableName, columnFamilyName);
                    }
                }
                List<HBaseColumn> columns = hBaseColumnFamily.getHBaseColumns();
                if (!CollectionUtils.isEmpty(columns)) {
                    String[] column = new String[columns.size()];
                    String[] values = new String[columns.size()];
                    int i = 0;
                    for (HBaseColumn colum : columns) {
                        if (StringUtils.hasText(colum.getValue())) {
                            values[i] = colum.getValue().toString();
                        } else {
                            values[i] = "";
                        }
                        column[i] = colum.getColumnName();
                        i++;
                    }
                    //进行相关列存储
                    insertRecords(tableName, hBaseRow.getRowName(), hBaseColumnFamily.getColumnFamilyName(), column, values);
                }
            }
        }
    }




    /**
     * @description 删除一行记录
     * @param tableName 表名
     * @param rowName 行名
     * @return void
     * @author abner chow
     * @date 2019/4/5 16:40
     */
    public void deleteRow(String tableName, String rowName) throws IOException {
        TableName name = TableName.valueOf(tableName);
        Table table = connection.getTable(name);
        Delete d = new Delete(rowName.getBytes());
        table.delete(d);
    }

    /**
     * @description  删除单行单列族记录
     * @param tableName 表名
     * @param rowName 行名
     * @param columnFamily 列族名
     * @return void
     * @author abner chow
     * @date 2019/4/5 16:42
     */
    public void deleteColumnFamily(String tableName, String rowName, String columnFamily) throws IOException {
        TableName name = TableName.valueOf(tableName);
        Table table = connection.getTable(name);
        Delete d = new Delete(rowName.getBytes()).addFamily(Bytes.toBytes(columnFamily));
        table.delete(d);
    }

    /**
     * @description 删除单行单列族单列记录
     * @param tableName 表名
     * @param rowName  行名
     * @param columnFamily  列族名
     * @param column  列名
     * @return void
     * @author abner chow
     * @date 2019/4/5 16:51
     */
    public void deleteColumn(String tableName, String rowName, String columnFamily, String column) throws IOException {
        TableName name = TableName.valueOf(tableName);
        Table table = connection.getTable(name);
        Delete d = new Delete(rowName.getBytes()).addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column));
        table.delete(d);
    }


    /**
     * @description 根据行获取行中的所有列族
     * @param tableName 表名
     * @param rowName 行名
     * @return java.util.Set
     * @author abner chow
     * @date 2019/4/12 16:05
     */
    public static Set getColumnFamilysByRow(String tableName, String rowName) throws IOException {
        Set result = new HashSet();
        TableName name = TableName.valueOf(tableName);
        Table table = connection.getTable(name);
        Get g = new Get(rowName.getBytes());
        Result rs = table.get(g);
        for (Cell cell : rs.rawCells()) {
            String family = Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
            result.add(family);
        }
        return result;
    }

    /**
     * @description 获取table的
     * @param tableName
     * @return java.util.Set
     * @author abner chow
     * @date 2019/4/12 18:21
     */
    public static Set getRowsByTableName(String tableName) throws IOException {
        Set resultSet = new HashSet();
        TableName name = TableName.valueOf(tableName);
        Table table = connection.getTable(name);
        Scan scan = new Scan();
        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            for (Cell cell : result.rawCells()) {
                String row = Bytes.toString(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
                resultSet.add(row);
            }
        }

        return resultSet;
    }



    /**
     * @description 查找单行单列族单列记录
     * @param tableName 表名
     * @param rowName 行名
     * @param columnFamily 列族名
     * @param column 列名
     * @return java.lang.String
     * @author abner chow
     * @date 2019/4/5 16:30
     */
    public static String getColumn(String tableName, String rowName, String columnFamily, String column) throws IOException {
        TableName name = TableName.valueOf(tableName);
        Table table = connection.getTable(name);
        Get g = new Get(rowName.getBytes());
        g.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column));
        Result rs = table.get(g);
        return Bytes.toString(rs.value());
    }

    /**
     * @description 根据 tablename, rowKey ,columnFamily获取记录
     * @param tableName 表名
     * @param rowName 行名
     * @param columnFamily 列族名
     * @return java.lang.String
     * @author abner chow
     * @date 2019/4/12 14:34
     */
    public static List<HBaseColumn> getByColumnFamily(String tableName, String rowName, String columnFamily) throws IOException {
        TableName name = TableName.valueOf(tableName);
        Table table = connection.getTable(name);
        Get g = new Get(rowName.getBytes());
        g.addFamily(Bytes.toBytes(columnFamily));
        Result rs = table.get(g);
        List<HBaseColumn> resultList = new ArrayList<HBaseColumn>();
        for (Cell cell : rs.rawCells()) {
            HBaseColumn hBaseColumn = new HBaseColumn();
            String colName = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
            String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
            hBaseColumn.setTableName(tableName);
            hBaseColumn.setRowName(rowName);
            hBaseColumn.setColumnFamilyName(columnFamily);
            hBaseColumn.setColumnName(colName);
            hBaseColumn.setValue(value);
            resultList.add(hBaseColumn);
        }
        return resultList;
    }


    /**
     * @description 查找一行记录
     * @param tableName 表名
     * @param rowName 行名
     * @return java.util.List<com.weiyu.hbase.entity.HBaseColumnFamily>
     * @author abner chow
     * @date 2019/4/5 17:50
     */
    public static List<HBaseColumnFamily> getByRow(String tableName, String rowName) throws IOException {

        List<HBaseColumnFamily> list = new ArrayList<HBaseColumnFamily>();
        Set columnFamilys = getColumnFamilysByRow(tableName, rowName);
        Iterator iterator = columnFamilys.iterator();
        if (iterator != null) {
            //遍历列族
            while (iterator.hasNext()) {
                //获取列族名
                String columnFamilyName = iterator.next().toString();
                HBaseColumnFamily hBaseColumnFamily = new HBaseColumnFamily();
                hBaseColumnFamily.setTableName(tableName);
                hBaseColumnFamily.setRowName(rowName);
                hBaseColumnFamily.setColumnFamilyName(columnFamilyName);
                //获取列族里的所有列
                List<HBaseColumn> columns = getByColumnFamily(tableName, rowName, columnFamilyName);
                hBaseColumnFamily.setHBaseColumns(columns);
                list.add(hBaseColumnFamily);
            }
        }
        return list;
    }

    /**
     * @description  根据表名获取记录
     * @param tableName
     * @return java.util.List<com.weiyu.hbase.entity.HBaseRow>
     * @author abner chow
     * @date 2019/4/12 18:12
     */
    public static List<HBaseRow> getByTable(String tableName) throws IOException {
        List<HBaseRow> result = new ArrayList<HBaseRow>();
        //获取表中所有的行row
        Set rowSet = getRowsByTableName(tableName);
        Iterator iterator = rowSet.iterator();
        if (iterator != null) {
            //遍历行
            while (iterator.hasNext()) {
                //获取行名
                String rowName = iterator.next().toString();
                HBaseRow hBaseRow = new HBaseRow();
                hBaseRow.setTableName(tableName);
                hBaseRow.setRowName(rowName);
                List<HBaseColumnFamily> familyList = new ArrayList<HBaseColumnFamily>();
                familyList = getByRow(tableName, rowName);
                hBaseRow.setHBaseColumnFamilies(familyList);
                result.add(hBaseRow);
            }
        }
        return result;
    }

    /**
     * 查询表中所有行（Scan方式）
     *
     * @param tablename
     * @return String
     */
    public String scanAllRecord(String tablename) throws IOException {
        String record = "";
        TableName name = TableName.valueOf(tablename);
        Table table = connection.getTable(name);
        Scan scan = new Scan();
        ResultScanner scanner = table.getScanner(scan);
        try {
            for (Result result : scanner) {
                for (Cell cell : result.rawCells()) {
                    String row = Bytes.toString(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
                    String family = Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
                    String colName = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
                    String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());

                    StringBuffer stringBuffer = new StringBuffer().append(row).append("\t")
                            .append(family).append("\t")
                            .append(colName).append("\t")
                            .append(value).append("\n");
                    String str = stringBuffer.toString();
                    record += str;
                }
            }
        } finally {
            if (scanner != null) {
                scanner.close();
            }
        }
        return record;
    }

    /**
     * 根据rowkey关键字查询报告记录
     *
     * @param tablename
     * @param rowKeyword
     * @return
     */
    public List scanReportDataByRowKeyword(String tablename, String rowKeyword) throws IOException {
        ArrayList<Object> list = new ArrayList<>();
        Table table = connection.getTable(TableName.valueOf(tablename));
        Scan scan = new Scan();
        //添加行键过滤器，根据关键字匹配
        RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, new SubstringComparator(rowKeyword));
        scan.setFilter(rowFilter);
        ResultScanner scanner = table.getScanner(scan);
        try {
            for (Result result : scanner) {
                //TODO 此处根据业务来自定义实现
                list.add(null);
            }
        } finally {
            if (scanner != null) {
                scanner.close();
            }
        }
        return list;
    }

    /**
     * 根据rowkey关键字和时间戳范围查询报告记录
     *
     * @param tablename
     * @param rowKeyword
     * @return
     */
    public List scanReportDataByRowKeywordTimestamp(String tablename, String rowKeyword, Long minStamp, Long maxStamp) throws IOException {
        ArrayList<Object> list = new ArrayList<>();
        Table table = connection.getTable(TableName.valueOf(tablename));
        Scan scan = new Scan();
        //添加scan的时间范围
        scan.setTimeRange(minStamp, maxStamp);
        RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, new SubstringComparator(rowKeyword));
        scan.setFilter(rowFilter);
        ResultScanner scanner = table.getScanner(scan);
        try {
            for (Result result : scanner) {
                //TODO 此处根据业务来自定义实现
                list.add(null);
            }
        } finally {
            if (scanner != null) {
                scanner.close();
            }
        }

        return list;
    }


    /**
     * @description 删除表操作
     * @param tableName
     * @return void
     * @author abner chow
     * @date 2019/4/5 17:47
     */
    public void deleteTable(String tableName) throws IOException {
        TableName name = TableName.valueOf(tableName);
        if (admin.tableExists(name)) {
            admin.disableTable(name);
            admin.deleteTable(name);
        }
    }

    /**
     * 利用协处理器进行全表count统计
     *
     * @param tablename
     */
    public Long countRowsWithCoprocessor(String tablename) throws Throwable {
        TableName name = TableName.valueOf(tablename);
        HTableDescriptor descriptor = admin.getTableDescriptor(name);
        String coprocessorClass = "org.apache.hadoop.hbase.coprocessor.AggregateImplementation";
        if (!descriptor.hasCoprocessor(coprocessorClass)) {
            admin.disableTable(name);
            descriptor.addCoprocessor(coprocessorClass);
            admin.modifyTable(name, descriptor);
            admin.enableTable(name);
        }
        //计时
        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
        Scan scan = new Scan();
        AggregationClient aggregationClient = new AggregationClient(conf);
        Long count = aggregationClient.rowCount(name, new LongColumnInterpreter(), scan);
        stopWatch.stop();
        System.out.println("RowCount：" + count + "，全表count统计耗时：" + stopWatch.getTotalTimeMillis());
        return count;
    }

}
